package cn.edu.laud.sany.utils;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

import cn.edu.laud.sany.dataimport.LaUDImport;
import cn.edu.laud.sany.types.ImportProcess;

public class ProcessManager implements Runnable
{
	public static long STATIC_BATCH_SIZE = 100000;
	public static long updateTimesToWriteFile = 100;
	private static MyLog logger;
	private static MyStatisticLog statisticLogger;
	private static MyErrorDataLog errorLogger;
	private static String csvFolder;
	private static String logFileFolder;
	private static String fileProcessMapFileName = "ProcessLog.txt";
	private static String fileProcessMapBackupFileName = "ProcessLog.txt.bak";
	private static String statisticLogFileName = "statisticLog.txt";
	private static String errorDataLogFileName = "errorDataLog.txt";
	private static HashMap<String, ImportProcess> fileProcessMap;
	private static int changeCountNum;
	private static long lastStatisticTime;
	public static long importStartTime;
	public static long midImportCount;
	public static long realImportCount;
	public static boolean writeProgressSpeedlog = true;
	public static boolean applicationExit = false;
	
	private static Integer requestUpdateCount = 0;
	
	private static ConcurrentLinkedQueue<ImportProcess> queue;
	
	public static void InitProcessManager(MyLog log, String csvfolder, String logfolder)
			throws IOException
	{
		logger = log;
		csvFolder = csvfolder;
		logFileFolder = logfolder;
		statisticLogger = MyStatisticLog.getMyStatisticLog(getFilePath(statisticLogFileName));
		statisticLogger.log("statistic logger start at [" + MyLog.getNowTimeString() + "].");
		errorLogger = MyErrorDataLog.getMyErrorDataLog(getFilePath(errorDataLogFileName));
		fileProcessMap = new HashMap<String, ImportProcess>();
		queue = new ConcurrentLinkedQueue<ImportProcess>();
		changeCountNum = 0;
		realImportCount = 0;
		midImportCount = 0;
		importStartTime = 0;
		lastStatisticTime = System.nanoTime();
		readProcessLog();
		updateQueue();
		Thread updateThread = new Thread(new ProcessManager());
		updateThread.start();
	}
	
	public static void CloseProcessManager()
	{
		saveProcessLog();
	}
	
	private static String getFilePath(String fileName)
	{
		return logFileFolder + "/" + fileName;
	}
	
	private static boolean readProcessLog()
	{
		File file = new File(getFilePath(fileProcessMapFileName));
		File fileBak = new File(getFilePath(fileProcessMapBackupFileName));
		if (!file.exists())
		{
			if (!fileBak.exists())
			{
				return false;
			}
			else
			{
				file = fileBak;
			}
		}
		synchronized (fileProcessMap)
		{
			try
			{
				BufferedReader sr1 = new BufferedReader(new FileReader(file));
				String content = null;
				while ((content = sr1.readLine()) != null)
				{
					try
					{
						ImportProcess process = ImportProcess.getImportProcess(content);
						if (process != null)
						{
							fileProcessMap.put(process.name, process);
						}
					}
					catch (Exception ex)
					{
						logger.error(ex.getMessage());
					}
				}
				sr1.close();
			}
			catch (FileNotFoundException e)
			{
				logger.info("Process log file not found, will create new.");
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}
		return true;
	}
	
	private static void saveProcessLog()
	{
		synchronized (fileProcessMap)
		{
			try
			{
				File file = new File(getFilePath(fileProcessMapFileName));
				File fileBak = new File(getFilePath(fileProcessMapBackupFileName));
				if (file.exists())
				{
					fileBak.delete();
					file.renameTo(fileBak);
				}
				BufferedWriter sw = new BufferedWriter(new FileWriter(
						getFilePath(fileProcessMapFileName), false));
				
				Iterator<ImportProcess> iter = fileProcessMap.values().iterator();
				while (iter.hasNext())
				{
					ImportProcess val = iter.next();
					sw.write(val.name);
					sw.write("\t");
					sw.write(String.valueOf(val.status));
					sw.write("\t");
					sw.write(String.valueOf(val.count));
					sw.write("\n");
				}
				sw.flush();
				sw.close();
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}
	}
	
	private static long getAllCount()
	{
		long allCount = 0;
		Iterator<ImportProcess> iter = fileProcessMap.values().iterator();
		while (iter.hasNext())
		{
			ImportProcess val = iter.next();
			allCount += val.count;
		}
		return allCount;
	}
	
	public static void updateQueue()
	{
		File[] files = getCSVFiles(new File(csvFolder));
		// 防止文件正在拷贝
		long size[] = new long[files.length];
		for (int i = 0; i < files.length; i++)
		{
			size[i] = files[i].length();
		}
		try
		{
			Thread.sleep(3000);
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
		for (int i = 0; i < files.length; i++)
		{
			if (size[i] == files[i].length())
			{
				String fileName = files[i].getAbsolutePath();
				ImportProcess process;
				if (fileProcessMap.containsKey(fileName))
				{
					process = fileProcessMap.get(fileName);
					if (process.status != 2 && process.isImporting == false
							&& queue.contains(process) == false)
					{
						queue.add(process);
					}
				}
				else
				{
					process = new ImportProcess(fileName, 0, 0);
					fileProcessMap.put(fileName, process);
					queue.add(process);
				}
			}
		}
	}
	
	private static File[] getCSVFiles(File file)
	{
		File[] files = new File[0];
		try
		{
			if (LaUDImport.csvCompressed)
			{
				files = file.listFiles(new FilenameFilter()
				{
					@Override
					public boolean accept(File dir, String name)
					{
						if (!name.endsWith("gz"))
							return false;
						else
							return true;
					}
				});
				
			}
			else
			{
				files = file.listFiles(new FilenameFilter()
				{
					@Override
					public boolean accept(File dir, String name)
					{
						if (!name.endsWith("csv"))
							return false;
						else
							return true;
					}
				});
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return files;
	}
	
	public static int getSourceCount()
	{
		return queue.size();
	}
	
	public static File getAvailableSource()
	{
		synchronized (queue)
		{
			if (queue.size() > 0)
			{
				ImportProcess process = queue.poll();
				process.status = 1;
				process.isImporting = true;
				File file = new File(process.name);
				return file;
			}
			else
			{
				requestUpdateQueue();
			}
		}
		return null;
	}
	
	public synchronized static void updateFileCount(String fileName, int status, long count)
	{
		if (fileProcessMap.containsKey(fileName))
		{
			ImportProcess process = fileProcessMap.get(fileName);
			process.status = status;
			process.count = count;
		}
		else
		{
			return;
		}
		changeCountNum++;
		
		if (changeCountNum >= updateTimesToWriteFile)
		{
			saveProcessLog();
			changeCountNum = 0;
		}
		
		if (writeProgressSpeedlog)
		{
			printProgressSpeedLog();
		}
	}
	
	private static void printProgressSpeedLog()
	{
		if (midImportCount >= STATIC_BATCH_SIZE)
		{			
			long allCount = getAllCount();
			long nowNanoSec = System.nanoTime();
			long allNanoSec = nowNanoSec - importStartTime;
			long oneTimeNanoSec = nowNanoSec - lastStatisticTime;
			lastStatisticTime = nowNanoSec;
			statisticLogger.log(String.valueOf(realImportCount));
			logger.info("Total import:" + allCount + " Real import: " + realImportCount
					+ " Time used:" + TimeUtils.nanoSecondToString(allNanoSec) + " Now speed: "
					+ (1000000000 * midImportCount / oneTimeNanoSec) + " Average speed: "
					+ (1000000000 * realImportCount / allNanoSec));
			midImportCount = 0;
		}
	}
	
	public static ImportProcess getProcessCount(String fileName)
	{
		if (fileProcessMap.containsKey(fileName))
		{
			return fileProcessMap.get(fileName);
		}
		return new ImportProcess(fileName, 0, 0);
	}
	
	public static boolean commitTask(String fileName, boolean succeed)
	{
		if (fileProcessMap.containsKey(fileName))
		{
			ImportProcess process = fileProcessMap.get(fileName);
			process.isImporting = false;
			if (succeed)
			{
				process.status = 2;
			}
			else
			{
				queue.add(process);
			}
			saveProcessLog();
			return true;
		}
		else
		{
			return false;
		}
	}
	
	public static void processTestLog(long count, long readtime, long importtime)
	{
		statisticLogger.log(String.format("%d %d %d", count, readtime, importtime));
	}
	
	public static void errorDataLog(String fileName, long lineNum, String data)
	{
		errorLogger.printLog(fileName, lineNum, data);
	}
	
	public static void dataTooManyLog(String fileName, long lineNum, String data)
	{
		errorLogger.printTooManyDataLog(fileName, lineNum, data);
	}
	
	public static synchronized void addRealImportCount(long count)
	{
		realImportCount += count;
		midImportCount += count;
	}
	
	public static void requestUpdateQueue()
	{
		synchronized (requestUpdateCount)
		{
			requestUpdateCount++;
		}
	}
	
	@Override
	public void run()
	{
		int emptyCount = 0;
		long lastUpdateTime = 0;
		while (applicationExit == false)
		{
			try
			{
				if (emptyCount > 100)
				{
					synchronized (requestUpdateCount)
					{
						requestUpdateCount++;
					}
					emptyCount = 0;
				}
				if (requestUpdateCount == 0)
				{
					emptyCount++;
				}
				else
				{
					long nowTime = System.currentTimeMillis() / 1000L;
					if (nowTime - lastUpdateTime > 10)
					{
						lastUpdateTime = System.currentTimeMillis() / 1000L;
						updateQueue();
						requestUpdateCount = 0;
					}
				}
				// sleep 1 sec
				Thread.sleep(1000);
			}
			catch (Exception e)
			{
			}
		}
	}
	
}
